/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.io.StringWriter;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;

import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.Reducer;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.Tuple;
import org.sss.mapreduce.datatype.DCPackableString;
import org.sss.mapreduce.datatype.PackableInt;
import org.sss.mapreduce.datatype.PackableJSONObject;
import org.sss.mapreduce.datatype.PackableString;
import org.sss.server.SssServer;

public class ContMergeReduceTest extends Reducer {
  private final Logger logger = SssServer.getLogger(ContMergeReduceTest.class);
  private int bufSize = 0;
  private int current = 0;
  private int scheduled = 0;      // # of scheduled
  private Object[] buffer = null;
  @Override
  public void configure(Context context) throws Exception {
    Configuration conf = context.getConfiguration();
    this.bufSize = conf.getInt("buffer.size");
    logger.info("buffer.size = {}", bufSize);
    this.current = 0;
    this.scheduled = 0;
    this.buffer = new Object[bufSize];
    for (int i=0; i<bufSize; i++) {
      buffer[i] = Collections.synchronizedList
        (new LinkedList<Tuple<PackableString, PackableJSONObject>>());
    }
  }
  @SuppressWarnings("unchecked")
  public void reduce(Context context,
                     DCPackableString key0, Iterable<PackableInt> values0,
                     DCPackableString key1, Iterable<PackableInt> values1,
                     Output<PackableString, PackableInt> output)
    throws Exception {
    logger.info("reduce method");
    String k0 = key0.get();
    String k1 = key1.get();
    assert k0.equals(k1): "key mismatch in merge reduce";
    int sum0 = 0;
    int sum1 = 0;
    for (PackableInt v: values0) {
      sum0 += v.get();
    }
    for (PackableInt v: values1) {
      sum1 += v.get();
    }
    List<Tuple<PackableString, PackableJSONObject>> currentBuffer =
      (List<Tuple<PackableString, PackableJSONObject>>) buffer[current];
    JSONObject obj = new JSONObject();
    obj.put("key", k0);
    obj.put("sum0", new Integer(sum0));
    obj.put("sum1", new Integer(sum1));
    obj.put("scheduled", new Integer(scheduled));
    StringWriter out = new StringWriter();
    obj.writeJSONString(out);
    String jsonText = out.toString();
    currentBuffer.add(new Tuple<PackableString, PackableJSONObject>
                      (new PackableString(k0), new PackableJSONObject(jsonText)));
  }
  @SuppressWarnings("unchecked")
  public void cleanup(Context context,
                      Output<PackableString, PackableJSONObject> output)
    throws SssException {
    logger.info("cleanup is called");
    synchronized (buffer) {
      for (int i=0; i<bufSize; i++) {
        for (Tuple<PackableString, PackableJSONObject> tuple:
               (List<Tuple<PackableString, PackableJSONObject>>) buffer[i]) {
          logger.info("calling output.write method");
          output.write(tuple.getKey(), tuple.getValue());
        }
      }
      current = (current + 1) % bufSize;
      ((List<Tuple<PackableString, PackableJSONObject>>) buffer[current]).clear();
    }
    scheduled++;
  }
}
